package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.bean.PaymentWide;
import com.atguigu.realtime.bean.ProductStats;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.common.GmallConstant;
import com.atguigu.realtime.util.DimQuery;
import com.atguigu.realtime.util.MyKafkaUtil;
import com.atguigu.realtime.util.MySinkUtil;
import com.atguigu.realtime.util.MyTimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.lang.reflect.InvocationTargetException;
import java.sql.Connection;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @ClassName: DWSProductStatsApp
 * @Description:
 * @Author: kele
 * @Date: 2021/4/26 11:08
 **/
public class DWSProductStatsApp extends BaseAppV2 {

    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        String[] topics = {
                ConstantTopic.DWD_DISPLAY,
                ConstantTopic.DWD_FAVOR_INFO,
                ConstantTopic.DWD_CART_INFO,
                ConstantTopic.DWM_ORDER_WIDE,
                ConstantTopic.DWM_PAYMENT_WIDE,
                ConstantTopic.DWD_ORDER_REFUND_INFO,
                ConstantTopic.DWD_COMMENT_INFO
        };

        new DWSProductStatsApp().init(40002, 2,
                "DWSProductStatsApp",
                "DWSProductStatsApp",
                ConstantTopic.DWD_PAGE,
                topics);
    }

    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStream<String>> streams) throws SQLException, ClassNotFoundException {


        DataStream<ProductStats> productStatsStream = UnionOneProductStream(streams);

        //  productStatsStream.print();

        //将数据聚合流
        SingleOutputStreamOperator<ProductStats> aggProductStream = aggregate(productStatsStream);

        //aggProductStream.print();

        //关联维度
        SingleOutputStreamOperator<ProductStats> addDimProductStream = joinDim(aggProductStream);

        //addDimProductStream.print();

        //写到clickhouse中
        send2Clickhouse(addDimProductStream);

        //将数据写入到kafka中
        send2Kafka(addDimProductStream);
    }

    private void send2Kafka(SingleOutputStreamOperator<ProductStats> stream) {

        stream
                .map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWS_PRODUCT_STATS));
    }

    //将数据写入到clickhouse中
    private void send2Clickhouse(SingleOutputStreamOperator<ProductStats> stream) {

        stream.addSink(MySinkUtil.getClickHouseSink("gmall2021", "product_stats_2021", ProductStats.class));
    }

    /**
     * 使用异步的方式关联
     *
     * @param aggProductStream
     */
    private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> aggProductStream) {

        return AsyncDataStream
                //无序的等待
                .unorderedWait(
                        aggProductStream,
                        new DimAsyncFunc<ProductStats>() {
                            /**
                             * 添加维度信息
                             * sku
                             */
                            @Override
                            protected void addDim(ProductStats input,
                                                  Connection connect,
                                                  Jedis redis,
                                                  ResultFuture resultFuture) {
                                try {

                                    //通过sku_info表添加
                                    JSONObject sku_info = DimQuery.getDimQuery(connect, redis, ConstantTopic.DIM_SKU_INFO, input.getSku_id());

                                    //添加sku信息
                                    input.setSku_name(sku_info.getString("SKU_NAME"));
                                    input.setSku_price(sku_info.getBigDecimal("PRICE"));

                                    //其他表查询时需要的数据
                                    input.setSpu_id(sku_info.getLong("SPU_ID"));
                                    input.setTm_id(sku_info.getLong("TM_ID"));
                                    input.setCategory3_id(sku_info.getLong("CATEGORY3_ID"));

                                    //添加spu信息
                                    JSONObject spu_info = DimQuery.getDimQuery(connect, redis, ConstantTopic.DIM_SPU_INFO, input.getSpu_id());
                                    input.setSpu_name(spu_info.getString("SPU_NAME"));

                                    //补齐品牌
                                    JSONObject tmInfo = DimQuery.getDimQuery(connect, redis, ConstantTopic.DIM_BASE_TRADEMARK, input.getTm_id());
                                    input.setTm_name(tmInfo.getString("TM_NAME"));

                                    //补齐三级品类
                                    JSONObject c3Info = DimQuery.getDimQuery(connect, redis, ConstantTopic.DIM_BASE_CATEGORY3, input.getCategory3_id());
                                    input.setCategory3_name(c3Info.getString("NAME"));

                                    resultFuture.complete(Collections.singletonList(input));

                                } catch (InvocationTargetException | IllegalAccessException | InstantiationException | SQLException e) {
                                    System.out.println("查询出错");
                                    e.printStackTrace();
                                }
                            }
                        },
                        30,
                        TimeUnit.SECONDS
                );
    }

    //按照商品进行聚合，得到每个商品的详细信息
    private SingleOutputStreamOperator<ProductStats> aggregate(DataStream<ProductStats> stream) {

        /**
         * 1、添加水印
         * 2、对商品进行keyBy
         * 3、将度量字段进行聚合
         */
        return stream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((ele, ts) -> ele.getTs())
                )
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<ProductStats>() {

                            //  将v2的值聚合到v1上，将v1输出

                            @Override
                            public ProductStats reduce(ProductStats v1,
                                                       ProductStats v2) throws Exception {
                                /**
                                 * 聚合字段：click_ct, display_ct, favor_ct, cart_ct,
                                 */
                                v1.setDisplay_ct(v1.getDisplay_ct() + v2.getDisplay_ct());
                                v1.setClick_ct(v1.getClick_ct() + v2.getClick_ct());
                                v1.setFavor_ct(v1.getFavor_ct() + v2.getFavor_ct());
                                v1.setCart_ct(v1.getCart_ct() + v2.getCart_ct());  //添加购物车
                                v1.setOrder_sku_num(v1.getOrder_sku_num() + v2.getOrder_sku_num());  //总的商品数（在订单宽表中添加的）

                                v1.setComment_ct(v1.getComment_ct() + v2.getComment_ct());
                                v1.setGood_comment_ct(v1.getGood_comment_ct() + v2.getGood_comment_ct());


                                v1.setOrder_amount(v1.getOrder_amount().add(v2.getOrder_amount()));         //订单宽表中添加的订单金额
                                v1.setPayment_amount(v1.getPayment_amount().add(v2.getPayment_amount()));  //支付总额（支付宽表 中添加的字段）
                                v1.setRefund_amount(v1.getRefund_amount().add(v2.getRefund_amount()));     //退款总额

                                //订单数，支付数，退款数应该从该窗口完成之后 从取集合的大小得到数量

                                v1.getOrderIdSet().addAll(v2.getOrderIdSet());
                                v1.getPaidOrderIdSet().addAll(v2.getPaidOrderIdSet());
                                v1.getRefundOrderIdSet().addAll(v2.getRefundOrderIdSet());

                                return v1;
                            }


                        },

                        /**
                         * 给字段设置值，
                         * 通过集合的大小 获取订单数，退款数，支付数
                         * 设置窗口的开始时间，结束时间
                         */
                        new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<ProductStats> elements,
                                                Collector<ProductStats> out) throws Exception {

                                ProductStats ps = elements.iterator().next();

                                //获取订单数，支付数，退单数
                                ps.setOrder_ct((long) ps.getRefundOrderIdSet().size());
                                ps.setPaid_order_ct((long) ps.getPaidOrderIdSet().size());
                                ps.setRefund_order_ct((long) ps.getRefundOrderIdSet().size());

                                ps.setStt(MyTimeUtil.DateTransfrom(ctx.window().getStart()));
                                ps.setEdt(MyTimeUtil.DateTransfrom(ctx.window().getEnd()));

                                out.collect(ps);

                            }
                        });
    }

    /**
     * 将有关商品的表进行合并，  最终通过sku_id进行keyby 和 ts进行window
     *
     * @param streams
     */
    private DataStream<ProductStats> UnionOneProductStream(Map<String, DataStream<String>> streams) {

        /**
         * 转化页面信息，添加 点击次数，
         *
         * 如何通过页面信息获取点击次数？
         * 获取点击，   通过判断页面信息中的 page_id，如果page_id的类型是good_detail，
         */
        SingleOutputStreamOperator<ProductStats> pageStream = streams
                .get(ConstantTopic.DWD_PAGE)
                .flatMap(new FlatMapFunction<String, ProductStats>() {
                    @Override
                    public void flatMap(String value,
                                        Collector<ProductStats> out) throws Exception {

                        JSONObject obj = JSON.parseObject(value);

                        JSONObject pageInfo = obj.getJSONObject("page");

                        if ("good_detail".equalsIgnoreCase(pageInfo.getString("page_id"))) {

                            ProductStats ps = new ProductStats();

                            ps.setSku_id(pageInfo.getLong("item"));   //添加商品编号
                            ps.setTs(obj.getLong("ts"));              //添加时间戳
                            ps.setClick_ct(1L);                           //将点击数置为1

                            out.collect(ps);
                        }

                    }
                });

        /**
         * 转化曝光流信息，转化为productState对象，添加商品编号，商品id、以及曝光次数
         */
        SingleOutputStreamOperator<ProductStats> displayStream = streams.get(ConstantTopic.DWD_DISPLAY)
                .flatMap(new FlatMapFunction<String, ProductStats>() {
                    @Override
                    public void flatMap(String value, Collector<ProductStats> out) throws Exception {

                        JSONObject displayInfo = JSON.parseObject(value);  //将曝光信息转化为json类型获取值

                        if ("sku_id".equalsIgnoreCase(displayInfo.getString("item_type"))) {

                            ProductStats ps = new ProductStats();

                            ps.setSku_id(displayInfo.getLong("item"));   //添加商品编号
                            ps.setTs(displayInfo.getLong("ts"));         //添加时间戳
                            ps.setDisplay_ct(1L);                            //将曝光数置为1

                            out.collect(ps);
                        }
                    }
                });

        /**
         * 获取收藏的信息
         */
        SingleOutputStreamOperator<ProductStats> favorStream = streams.get(ConstantTopic.DWD_FAVOR_INFO)
                .map(value -> {

                    JSONObject obj = JSON.parseObject(value);

                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(MyTimeUtil.TransfromTs(obj.getString("create_time")));
                    ps.setFavor_ct(1L);

                    return ps;
                });

        SingleOutputStreamOperator<ProductStats> cartStream = streams
                .get(ConstantTopic.DWD_CART_INFO)
                .map(value -> {

                    JSONObject obj = JSON.parseObject(value);

                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(MyTimeUtil.TransfromTs(obj.getString("create_time")));
                    ps.setCart_ct(1L);

                    return ps;
                });

        /**
         * 转化订单宽表，增加：sku编号，商品价格，订单id，
         */
        SingleOutputStreamOperator<ProductStats> orderWideStream = streams
                .get(ConstantTopic.DWM_ORDER_WIDE)
                .map(value -> {

                    OrderWide obj = JSON.parseObject(value, OrderWide.class);

                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getSku_id());
                    ps.setTs(MyTimeUtil.TransfromTs(obj.getCreate_time()));

                    //该字段通过后面聚合的时候 获取getOrderIdSet的长度
                    // ps.setOrder_ct(1L);

                    ps.setOrder_amount(obj.getSplit_total_amount());  //获取每件商品的价格
                    ps.setOrder_sku_num(obj.getSku_num());            //获取sku数目

                    ps.getOrderIdSet().add(obj.getOrder_id());        //获取订单id

                    return ps;
                });

        /**
         * 添加支付宽表信息，增加：sku_id，每件商品的支付金额，每件商品的订单号放入集合中
         */
        SingleOutputStreamOperator<ProductStats> paymentStream = streams
                .get(ConstantTopic.DWM_PAYMENT_WIDE)
                .map(new MapFunction<String, ProductStats>() {

                    @Override
                    public ProductStats map(String value) throws Exception {
                        PaymentWide obj = JSON.parseObject(value, PaymentWide.class);

                        ProductStats ps = new ProductStats();

                        ps.setSku_id(obj.getSku_id());
                        ps.setTs(MyTimeUtil.TransfromTs(obj.getPayment_create_time()));
                        ps.setPayment_amount(obj.getSplit_total_amount());

                        ps.getPaidOrderIdSet().add(obj.getOrder_id()); //将订单id添加到支付订单信息中
                        return ps;
                    }
                });

        /**
         * 添加退款信息
         * 增加：
         */
        SingleOutputStreamOperator<ProductStats> refundStream = streams
                .get(ConstantTopic.DWD_ORDER_REFUND_INFO)
                .map(new MapFunction<String, ProductStats>() {

                    @Override
                    public ProductStats map(String value) throws Exception {

                        JSONObject obj = JSON.parseObject(value);

                        ProductStats ps = new ProductStats();

                        ps.setSku_id(obj.getLong("sku_id"));
                        ps.setTs(MyTimeUtil.TransfromTs(obj.getString("create_time")));
                        //统计退款金额
                        ps.setRefund_amount(obj.getBigDecimal("refund_amount"));
                        //将退款的金额放入集合中返回
                        ps.getRefundOrderIdSet().add(obj.getLong("order_id"));

                        return ps;
                    }
                });

        SingleOutputStreamOperator<ProductStats> commentStream = streams.get(ConstantTopic.DWD_COMMENT_INFO)
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);

                        ProductStats ps = new ProductStats();

                        ps.setSku_id(obj.getLong("sku_id"));
                        ps.setTs(MyTimeUtil.TransfromTs(obj.getString("create_time")));
                        ps.setComment_ct(1L);


                        if (GmallConstant.APPRAISE_GOOD.equalsIgnoreCase(obj.getString("appraise"))) {
                            ps.setGood_comment_ct(1L);
                        }

                        return ps;

                    }
                });

        return pageStream.union(displayStream,
                favorStream,
                cartStream,
                orderWideStream,
                paymentStream,
                refundStream,
                commentStream);
    }
}

